package com.shujia.hbase.javaapi;

import com.shujia.hdfs.Student;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;

public class Index {
    private HConnection connection;
    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {
        /**
         * 通过java代码操作hbase
         *
         * 1、建立连接
         * 2、对数据和 表进行操作
         * 3、关闭连接
         *
         */

        //创建hadoop配置文件对象
        Configuration conf = new Configuration();

        //指定zk连接地址
        conf.set("hbase.zookeeper.quorum", "node2:2181,node3:2181.node4:2181");

        //创建连接

        /**
         * 对表进行操作连接maste
         *
         * 对数据进行操作连接regionserver
         *
         */

        try {
            //增删改数据使用
            connection = HConnectionManager.createConnection(conf);

            //对表操作时使用
            hBaseAdmin = new HBaseAdmin(conf);
            System.out.println("连接创建成功");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void queryClazz() throws Exception {

        long start = System.currentTimeMillis();

        String clazz = "文科一班";

        //查询班级所有学生
        HTableInterface student = connection.getTable("student");

        //前缀比较器
        BinaryPrefixComparator prefixComparator = new BinaryPrefixComparator(clazz.getBytes());


        SingleColumnValueFilter filter = new SingleColumnValueFilter(
                "info".getBytes(),//列簇
                "clazz".getBytes(),//列名
                CompareFilter.CompareOp.EQUAL,//比较规整
                prefixComparator//比较器
        );

        Scan scan = new Scan();

        //增加过滤器
        scan.setFilter(filter);

        ResultScanner results = student.getScanner(scan);

        //打印结果
        print(results);

        long end = System.currentTimeMillis();

        System.out.println(end - start);


    }


    @Test
    public void queryClazzIndex() throws Exception {

        long start = System.currentTimeMillis();
        /**
         * 先查询索引表
         * 再 查询数据表
         *
         */

        String clazz = "文科一班";

        HTableInterface student_index = connection.getTable("student_index");

        Scan scan = new Scan();
        //前缀比较器
        BinaryPrefixComparator prefixComparator = new BinaryPrefixComparator(clazz.getBytes());

        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, prefixComparator);

        scan.setFilter(rowFilter);

        ResultScanner scanner = student_index.getScanner(scan);

        ArrayList<Get> gets = new ArrayList<>();


        for (Result result : scanner) {
            String rowkey = Bytes.toString(result.getRow());

            String id = rowkey.split("-")[1];

            Get get = new Get(id.getBytes());

            gets.add(get);
        }

        //查询数据表
        HTableInterface student = connection.getTable("student");

        Result[] results = student.get(gets);

        print(results);


        long end = System.currentTimeMillis();

        System.out.println(end - start);

    }


    @Test
    public void insertIndex() {
        Student student = new Student();
        student.setId("1500101001");
        student.setName("于向南");
        student.setAge(24);
        student.setGender("男");
        student.setClazz("文科一班");


        /**
         * 插入数据
         * 1、将索引写入索引表
         * 2、将数据写入数据表
         *
         */
        //1、插入索引
        try {

            HTableInterface student_index = connection.getTable("student_index");

            String indexRowkey = student.getClazz() + "-" + student.getId();

            Put put = new Put(indexRowkey.getBytes());
            put.add("i".getBytes(), "n".getBytes(), "".getBytes());

            student_index.put(put);


        } catch (IOException e) {
            e.printStackTrace();
            return;
        }

        //2,插入数据
        try {
            HTableInterface student1 = connection.getTable("student");
            Put put = new Put(student.getId().getBytes());

            put.add("info".getBytes(), "name".getBytes(), student.getName().getBytes());
            put.add("info".getBytes(), "age".getBytes(), student.getAge().toString().getBytes());
            put.add("info".getBytes(), "gender".getBytes(), student.getGender().getBytes());
            put.add("info".getBytes(), "clazz".getBytes(), student.getClazz().getBytes());

            student1.put(put);


        } catch (IOException e) {
            e.printStackTrace();

            //如果插入数据失败，删除索引表中的索引
            try {
                HTableInterface student_index = connection.getTable("student_index");
                String indexRowkey = student.getClazz() + "-" + student.getId();
                Delete delete = new Delete(indexRowkey.getBytes());
                student_index.delete(delete);
            } catch (IOException e1) {
                e1.printStackTrace();
            }

        }
    }

    @Test
    public void deleteIndex() {

        String id = "1500101001";


        /**
         * 删除数据
         * 1、删除索引表
         * 2、删除数据
         *
         * 如果删除索引失败，不在删除数据
         * 如果删除数据失败需要恢复以及删除的索引
         *
         */

        String indexRowkey = "";
        try {

            //通过学号查询学生的班级
            HTableInterface student = connection.getTable("student");
            Result result = student.get(new Get(id.getBytes()));
            if (result.isEmpty()) {
                return;
            }

            byte[] value = result.getValue("info".getBytes(), "clazz".getBytes());
            String clazz = Bytes.toString(value);


            indexRowkey = clazz + "-" + id;

            HTableInterface student_index = connection.getTable("student_index");
            Delete delete = new Delete(indexRowkey.getBytes());
            //删除索引
            student_index.delete(delete);

        } catch (IOException e) {
            e.printStackTrace();
            return;
        }

        //删除数据表
        try {
            HTableInterface student = connection.getTable("student");
            Delete delete = new Delete(id.getBytes());
            student.delete(delete);

        } catch (IOException e) {
            e.printStackTrace();

            //如果删除数据失败，需要还原索引表数据

            try {
                HTableInterface student_index = connection.getTable("student_index");
                Put put = new Put(indexRowkey.getBytes());
                put.add("i".getBytes(), "n".getBytes(), "".getBytes());
                student_index.put(put);

            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }

    @Test
    public void insert() {
        Student student = new Student();
        student.setId("1500101001");
        student.setName("于向南");
        student.setAge(24);
        student.setGender("男");
        student.setClazz("文科一班");


        //2,插入数据
        try {
            HTableInterface student1 = connection.getTable("student");
            Put put = new Put(student.getId().getBytes());

            put.add("info".getBytes(), "name".getBytes(), student.getName().getBytes());
            put.add("info".getBytes(), "age".getBytes(), student.getAge().toString().getBytes());
            put.add("info".getBytes(), "gender".getBytes(), student.getGender().getBytes());
            put.add("info".getBytes(), "clazz".getBytes(), student.getClazz().getBytes());

            student1.put(put);


        } catch (IOException e) {
        }
    }


    public void print(ResultScanner results) {
        for (Result result : results) {
            //取出rowkey
            String id = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }
    }

    public void print(Result[] results) {
        for (Result result : results) {
            //取出rowkey
            String id = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }
    }


    @After
    public void close() {

        try {
            hBaseAdmin.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
